package com.linx.rxlife

import androidx.lifecycle.Lifecycle
import androidx.lifecycle.LifecycleOwner
import io.reactivex.functions.Action
import io.reactivex.functions.Consumer
import io.reactivex.internal.functions.Functions
import io.reactivex.internal.subscribers.LambdaSubscriber
import org.reactivestreams.Publisher
import org.reactivestreams.Subscription

class FlowableLife<T>(
    private val source: Publisher<T>,
    owner: LifecycleOwner,
    atLeastState: Lifecycle.State = Lifecycle.State.STARTED
) : RxLife(owner, atLeastState) {

    private lateinit var onNext: Consumer<in T>
    private lateinit var onError: Consumer<in Throwable>
    private lateinit var onComplete: Action
    private lateinit var onSubscribe: Consumer<in Subscription>

    override fun onActive() {
        val subscriber = LambdaSubscriber<T>(
            onNext, onError, onComplete, onSubscribe
        )
        source.subscribe(subscriber)
        disposable = subscriber
    }

    fun observe(
        onNext: Consumer<in T>,
        onError: Consumer<in Throwable> = Functions.ON_ERROR_MISSING,
        onComplete: Action = Functions.EMPTY_ACTION,
        onSubscribe: Consumer<in Subscription> = Functions.emptyConsumer()
    ) {
        if (owner.lifecycle.currentState != Lifecycle.State.DESTROYED) {
            owner.lifecycle.addObserver(this)
            this.onNext = onNext
            this.onError = onError
            this.onComplete = onComplete
            this.onSubscribe = onSubscribe
            activeStateChange(shouldBeActive())
        }
    }

}